package com.spider.moudle.util;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.common.security.JaasUtils;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;

public class KafkaUtil {


	/**
	 * 创建ZKUtils
	 */
	private static ZkUtils build() {
		return ZkUtils.apply(PropertiesUtil.getVal("kafka.zookeeper.host") + ":2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
	}

	public static Boolean topicExist(String topic) {
		ZkUtils zkUtils = build();
		Boolean flag = AdminUtils.topicExists(zkUtils, topic);
		zkUtils.close();
		return flag;
	}

	public static void topicCreate(String topic) {
		ZkUtils zkUtils = build();
		AdminUtils.createTopic(zkUtils, topic, Integer.parseInt(PropertiesUtil.getVal("kafka.partition.size")), 1, new Properties(), RackAwareMode.Enforced$.MODULE$);
		zkUtils.close();
	}

	public static void topicDelete(String topic) {
		ZkUtils zkUtils = build();
		AdminUtils.deleteTopic(zkUtils, topic);   //删除队列
		zkUtils.close();
	}


	/**
	 * 检查所有的topic是否存在,如果不存在则创建该topic
	 */
	public static void topicCheck() throws Exception {
		List<String> topic_list = new ArrayList<>();
		topic_list.add(PropertiesUtil.getVal("kafka.speech.data.topic"));
		topic_list.add(PropertiesUtil.getVal("kafka.user.data.topic"));
		topic_list.add(PropertiesUtil.getVal("kafka.speech.task.topic"));
		topic_list.add(PropertiesUtil.getVal("kafka.user.task.topic"));
		for(String topic : topic_list) {
			if(!KafkaUtil.topicExist(topic)) {
				KafkaUtil.topicCreate(topic);
				System.err.println("topic: " + topic + " 不存在, 创建该topic");
			}
		}
	}

    public static void main(String[] args) {
    	String topic="1101_SPIDER_USER";
    	Boolean isExist = KafkaUtil.topicExist(topic);
    	if(isExist == null) {
    		return;
		}
    	if(!isExist) {
			System.out.println("topic: " + topic + " 不存在");
			KafkaUtil.topicCreate(topic);
			System.out.println("topic: " + topic + " 创建成功");
		} else {
			System.out.println("topic: " + topic + " 存在");
			KafkaUtil.topicDelete(topic);
			System.out.println("topic: " +topic + " 删除成功");
		}
		isExist = KafkaUtil.topicExist(topic);
		if(!isExist) {
			System.out.println("topic: " + topic + " 不存在");
		} else {
			System.out.println("topic: " + topic + " 存在");
		}
     }
}
